package com.it.dao;

import com.it.constants.Constants;
import com.it.utils.HBaseUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.logging.Logger;

/**
 * 1.发布微博
 * 2.删除微博
 * 3.关注用户
 * 4.取关用户
 * 5.获取微博详情
 * 6.获取用户的初始化页面
 *
 * @author : code1997
 * @date :2021-02-2021/2/4 21:54
 */
public class HBaseDao {

    private static Logger logger = Logger.getLogger(HBaseUtils.class.getName());

    /**
     * first,add record to weibo:content table;
     * second,get fans from weibo:relation
     */
    public static void publishWeiBo(String uid, String content) {
        Connection connection = null;
        Table contentTable = null;
        Table relationTable = null;
        Table inboxTable = null;
        try {
            connection = ConnectionFactory.createConnection();
            contentTable = connection.getTable(TableName.valueOf(Constants.TABLE_CONTENT));
            connection.getTable(TableName.valueOf(Constants.TABLE_CONTENT));
            long ts = System.currentTimeMillis();
            String rowKey = uid + "_" + (Constants.MAX_TIME - ts);
            Put contentPut = new Put(Bytes.toBytes(rowKey));
            contentPut.addColumn(
                    Bytes.toBytes(Constants.TABLE_CONTENT_CF),
                    Bytes.toBytes("content"),
                    Bytes.toBytes(content));
            contentTable.put(contentPut);
            relationTable = connection.getTable(TableName.valueOf(Constants.TABLE_RELATIONS));
            Get fanGet = new Get(Bytes.toBytes(uid));
            fanGet.addFamily(Bytes.toBytes(Constants.TABLE_RELATIONS_CF2));
            Result result = relationTable.get(fanGet);
            //inbox record list
            ArrayList<Put> inboxPuts = new ArrayList<>();
            for (Cell cell : result.rawCells()) {
                Put inboxPut = new Put(CellUtil.cloneQualifier(cell));
                inboxPut.addColumn(Bytes.toBytes(Constants.TABLE_INBOX_CF),
                        Bytes.toBytes(uid),
                        Bytes.toBytes(rowKey));
                inboxPuts.add(inboxPut);
            }
            if (inboxPuts.size() > 0) {
                inboxTable = connection.getTable(TableName.valueOf(Constants.TABLE_INBOX));
                inboxTable.put(inboxPuts);
            } else {
                System.out.println("该用户暂时没有粉丝!");
            }
        } catch (IOException e) {
            System.out.println(e.getMessage());
        } finally {
            try {
                HBaseUtils.closeResource(null, connection, contentTable, relationTable, inboxTable);
            } catch (IOException e) {
                System.out.println(e.getMessage());
            }
        }
    }

    /**
     * first:modify relation table
     */
    public static void addAttends(String uid, String... attends) {
        if (attends.length <= 0) {
            System.out.println("请选择待选择的人！");
            return;
        }
        Connection connection = null;
        Table contentTable = null;
        Table relationTable = null;
        Table inboxTable = null;
        try {
            //first part
            connection = ConnectionFactory.createConnection();
            relationTable = connection.getTable(TableName.valueOf(Constants.TABLE_RELATIONS));
            ArrayList<Put> relationPuts = new ArrayList<>();
            Put uidPut = new Put(Bytes.toBytes(uid));
            for (String attend : attends) {
                uidPut.addColumn(Bytes.toBytes(Constants.TABLE_RELATIONS_CF1),
                        Bytes.toBytes(attend),
                        Bytes.toBytes(attend));
                Put attendPut = new Put(Bytes.toBytes(attend));
                attendPut.addColumn(Bytes.toBytes(Constants.TABLE_RELATIONS_CF2),
                        Bytes.toBytes(uid),
                        Bytes.toBytes(uid));
                relationPuts.add(attendPut);
            }
            relationPuts.add(uidPut);
            relationTable.put(relationPuts);

            //second part
            contentTable = connection.getTable(TableName.valueOf(Constants.TABLE_CONTENT));
            Put inboxPut = new Put(Bytes.toBytes(uid));
            for (String attend : attends) {
                Scan scan = new Scan(Bytes.toBytes(attend + "_"),
                        Bytes.toBytes(attend + "|"));
                ResultScanner contenteScanner = contentTable.getScanner(scan);
                //存在问题1：如果有人发布微博比较多，那么我们可能会导致加载过多的无用数据，我们可以存储的时候直接进行反转。
                //存在问题2：防止数据同时加载导致数据的时间戳一致，导致只可以返回一条数据。
                long ts = System.currentTimeMillis();
                for (Result result : contenteScanner) {
                    inboxPut.addColumn(Bytes.toBytes(Constants.TABLE_INBOX_CF),
                            Bytes.toBytes(attend),
                            ts++,
                            result.getRow());
                }
            }
            if (!inboxPut.isEmpty()) {
                inboxTable = connection.getTable(TableName.valueOf(Constants.TABLE_INBOX));
                inboxTable.put(inboxPut);
            } else {
                System.out.println("暂无微博，可以关注其他人");
            }


        } catch (IOException e) {
            System.out.println(e.getMessage());
        } finally {
            //final park
            try {
                HBaseUtils.closeResource(null, connection, contentTable, relationTable, inboxTable);
            } catch (IOException e) {
                System.out.println(e.getMessage());
            }
        }
    }

    public static void delAttends(String uid, String... dels) {
        if (dels.length <= 0) {
            System.out.println("请选择需要取消关注的用户！");
        }
        Connection connection = null;
        Table contentTable = null;
        Table relationTable = null;
        Table inboxTable = null;
        try {
            connection = ConnectionFactory.createConnection(Constants.CONFIGURATION);
            relationTable = connection.getTable(TableName.valueOf(Constants.TABLE_RELATIONS));
            ArrayList<Delete> delList = new ArrayList<>();
            Delete uidDelete = new Delete(Bytes.toBytes(uid));
            for (String del : dels) {
                uidDelete.addColumns(Bytes.toBytes(Constants.TABLE_RELATIONS_CF1), Bytes.toBytes(del));
                Delete delete = new Delete(Bytes.toBytes(del));
                delete.addColumns(Bytes.toBytes(Constants.TABLE_RELATIONS_CF2), Bytes.toBytes(uid));
                delList.add(delete);
            }
            delList.add(uidDelete);
            relationTable.delete(delList);
            inboxTable = connection.getTable(TableName.valueOf(Constants.TABLE_INBOX));
            Delete inboxDel = new Delete(Bytes.toBytes(uid));
            for (String del : dels) {
                inboxDel.addColumns(Bytes.toBytes(Constants.TABLE_INBOX_CF), Bytes.toBytes(del));
            }

            inboxTable.delete(inboxDel);

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            //final park
            try {
                HBaseUtils.closeResource(null, connection, contentTable, relationTable, inboxTable);
            } catch (IOException e) {
                System.out.println(e.getMessage());
            }
        }

    }

    public static void getInit(String uid) {

        Connection connection = null;
        Table contentTable = null;
        Table relationTable = null;
        Table inboxTable = null;
        try {
            connection = ConnectionFactory.createConnection(Constants.CONFIGURATION);
            inboxTable = connection.getTable(TableName.valueOf(Constants.TABLE_INBOX));
            contentTable = connection.getTable(TableName.valueOf(Constants.TABLE_CONTENT));
            Get get = new Get(Bytes.toBytes(uid));
            get.setMaxVersions(2);
            Result initResult = inboxTable.get(get);
            logger.info("关注的人数---->" + initResult.rawCells().length);
            for (Cell cell : initResult.rawCells()) {
                logger.info("关注的信息rowKey---->" + cell.toString());
                Get contGet = new Get(CellUtil.cloneValue(cell));
                Result contentResult = contentTable.get(contGet);
                for (Cell rawCell : contentResult.rawCells()) {
                    System.out.println("rk==>" + Bytes.toString(rawCell.getRow()) + "\t"
                            + "CF==>" + Bytes.toString(rawCell.getFamily()) + "\t"
                            + "CN==>" + Bytes.toString(rawCell.getQualifier()) + "\t"
                            + "VA==>" + Bytes.toString(rawCell.getValue())
                    );
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            //final park
            try {
                HBaseUtils.closeResource(null, connection, contentTable, relationTable, inboxTable);
            } catch (IOException e) {
                System.out.println(e.getMessage());
            }
        }
    }

    /**
     * 简化：只获取指定uid的微博即可
     */
    public static void getWeiBo(String uid) {
        Connection connection = null;
        Table contentTable = null;
        Table relationTable = null;
        Table inboxTable = null;
        try {
            connection = ConnectionFactory.createConnection(Constants.CONFIGURATION);
            contentTable = connection.getTable(TableName.valueOf(Constants.TABLE_CONTENT));
            Scan scan = new Scan();
            Filter filter = new RowFilter(CompareFilter.CompareOp.GREATER_OR_EQUAL, new SubstringComparator(uid + "_"));
            scan.setFilter(filter);
            ResultScanner scanner = contentTable.getScanner(scan);
            for (Result result : scanner) {
                logger.info(uid + "的微博个数为:" + result.rawCells().length);
                for (Cell rawCell : result.rawCells()) {
                    System.out.println("rk==>" + Bytes.toString(rawCell.getRow()) + "\t"
                            + "CF==>" + Bytes.toString(rawCell.getFamily()) + "\t"
                            + "CN==>" + Bytes.toString(rawCell.getQualifier()) + "\t"
                            + "VA==>" + Bytes.toString(rawCell.getValue())
                    );
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            //final park
            try {
                HBaseUtils.closeResource(null, connection, contentTable, relationTable, inboxTable);
            } catch (IOException e) {
                System.out.println(e.getMessage());
            }
        }


    }

}
